﻿using System;
using System.Diagnostics.Contracts;
using System.IO;
using System.Threading;
using System.Transactions;
using O1.Data;
using O1.Data.Traversal;

namespace O1.IO.Storage
{
    [ContractVerification(true)]
    public class MemoryTrunk
    {
        private const int PaddingSize = 1024;

        private const int EstimatedStorageOverhead = 8;

        private readonly ushort trunkId;

        private readonly StorageBlock storage;

        private readonly TransactionLog log;

        private long estimatedSizeInBytes;

        private long maxAllowedSizeInBytes;

        private long lockValue;

        public MemoryTrunk(int trunkId, int initialCapacity, long maxAllowedSizeInBytes, TransactionLog log)
            : this(trunkId, initialCapacity, log)
        {
            Contract.Requires<ArgumentNullException>(log != null);
            Contract.Requires<ArgumentOutOfRangeException>(trunkId >= 0 && trunkId <= ushort.MaxValue);
            Contract.Requires<ArgumentOutOfRangeException>(maxAllowedSizeInBytes > (initialCapacity * EstimatedStorageOverhead));

            this.maxAllowedSizeInBytes = maxAllowedSizeInBytes;
        }

        private MemoryTrunk(int trunkId, int initialCapacity, TransactionLog log)
        {
            Contract.Requires(log != null);
            Contract.Requires(trunkId >= 0 && trunkId <= ushort.MaxValue);

            this.storage = new StorageBlock(initialCapacity);
            this.trunkId = (ushort)trunkId;
            this.log = log;
        }

        public int TrunkId
        {
            get
            {
                return this.trunkId;
            }
        }

        public long Capacity
        {
            get
            {
                return this.storage.Length;
            }
        }

        public static MemoryTrunk Load(PersistentTrunk storage, decimal growthFactor, TransactionLog log)
        {
            Contract.Requires<ArgumentNullException>(log != null);
            Contract.Requires<ArgumentNullException>(storage != null);
            Contract.Requires<ArgumentNullException>(storage.Primary != null);
            Contract.Requires<ArgumentNullException>(storage.Primary.Stream != null);
            Contract.Requires<ArgumentOutOfRangeException>(storage.CellCount > 0);
            Contract.Requires<ArgumentOutOfRangeException>(growthFactor > 0 && growthFactor < 1);
            
            Contract.Assert(storage.TrunkId >= 0 && storage.TrunkId <= ushort.MaxValue);
            var trunk = new MemoryTrunk(storage.TrunkId, storage.CellCount, log);
            trunk.estimatedSizeInBytes = storage.SizeInBytes + (storage.CellCount * EstimatedStorageOverhead);
            trunk.maxAllowedSizeInBytes = trunk.estimatedSizeInBytes + (int)(trunk.estimatedSizeInBytes * growthFactor);
            using (var adapter = new InputStreamAdapter(storage.Primary.Stream, storage.Primary.FileSize, false))
            {
                ReadInput(trunk, adapter);
            }

            var differences = storage.Differences();
            if (differences != null)
            {
                foreach (var file in differences)
                {
                    if (file != null && file.Stream != null)
                    {
                        using (var adapter = new InputStreamAdapter(file.Stream, file.FileSize, false))
                        {
                            ReadInput(trunk, adapter);
                        }
                    }
                }
            }

            return trunk;
        }

        [Pure]
        public bool HasCapacity(int nextSize)
        {
            return this.estimatedSizeInBytes + nextSize <= this.maxAllowedSizeInBytes;
        }

        public void AddCell(long cellId, byte[] data)
        {
            Contract.Requires<ArgumentNullException>(data != null);
            Contract.Requires<ArgumentException>(data.Length > 3);
            Contract.Requires<InvalidOperationException>(this.HasCapacity(data.Length));
            Contract.Ensures(this.Capacity > cellId);

            this.CheckAllocation(cellId);
            var spinLock = new ConcurrentWriter();
            using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew))
            {
                var instance = new TransactionLogEnlistment(this.log, data);
                Contract.Assume(Transaction.Current != null);
                Transaction.Current.EnlistVolatile(spinLock, EnlistmentOptions.EnlistDuringPrepareRequired);
                Transaction.Current.EnlistVolatile(instance, EnlistmentOptions.EnlistDuringPrepareRequired);

                var previous = spinLock.SpinWrite(data);
                this.storage[cellId] = spinLock.DirtyRead();
                var previousSize = (previous != null ? previous.Length : 0) + EstimatedStorageOverhead;
                this.estimatedSizeInBytes += (0 - previousSize) + data.Length + EstimatedStorageOverhead;
                scope.Complete();
            }
        }

        public void SetCell(long cellId, byte[] data)
        {
            Contract.Requires<ArgumentNullException>(data != null);
            Contract.Requires<ArgumentOutOfRangeException>(cellId < this.Capacity);

            var spinLock = new ConcurrentWriter();
            using (var scope = new TransactionScope(TransactionScopeOption.RequiresNew))
            {
                var instance = new TransactionLogEnlistment(this.log, data);
                Contract.Assume(Transaction.Current != null);
                Transaction.Current.EnlistVolatile(spinLock, EnlistmentOptions.EnlistDuringPrepareRequired);
                Transaction.Current.EnlistVolatile(instance, EnlistmentOptions.EnlistDuringPrepareRequired);

                var previous = spinLock.SpinWrite(data);
                var previousSize = (previous != null ? previous.Length : 0) + EstimatedStorageOverhead;

                this.estimatedSizeInBytes += (0 - previousSize) + data.Length + EstimatedStorageOverhead;
                scope.Complete();
            }
        }

        public byte[] GetCell(long cellId)
        {
            if (cellId >= this.storage.Length)
            {
                return null;
            }

            return this.storage[cellId];
        }

        public void MakeReadOnly()
        {
            this.maxAllowedSizeInBytes = this.estimatedSizeInBytes;
        }

        private static void ReadInput(MemoryTrunk trunk, InputAdapter adapter)
        {
            Contract.Requires(trunk != null);
            Contract.Requires(adapter != null);

            while (!adapter.EndOfStream)
            {
                // Get the size of the cell first:
                var offset = adapter.PopOffset(Sz.CLR.Int32);
                if (offset > 0)
                {
                    Contract.Assert(adapter.SharedBuffer.Length > offset + Sz.CLR.Int32);
                    var size = Math.Max(adapter.SharedBuffer.ToInt32(offset), Sz.CLR.Int32);
                    
                    Contract.Assert(size >= Sz.CLR.Int32);
                    if (adapter.PeekAhead(size) != Markup.CommittedVertex)
                    {
                        continue;
                    }

                    byte[] cell;
                    if (adapter.TryPopOffset(size, out offset))
                    {
                        Contract.Assume(offset >= 0);
                        Contract.Assume(offset + size <= adapter.SharedBuffer.Length);
                        cell = ReadWholeCell(adapter, offset, size);
                    }
                    else
                    {
                        cell = IterateCell(adapter, size, (ushort)trunk.TrunkId);
                    }

                    Contract.Assume(cell.Length >= Sz.CLR.Int32);
                    var cellId = cell.ToUInt32(0);

                    Contract.Assume(trunk.HasCapacity(cell.Length));
                    trunk.CheckAllocation(cellId);

                    Contract.Assert(cellId <= trunk.storage.Length);
                    trunk.SetCell(cellId, cell);
                }
            }
        }

        private static byte[] IterateCell(InputAdapter adapter, int size, ushort trunkId)
        {
            Contract.Requires(adapter != null);

            var builder = new CellBuilder(size);
            var observer = new CellObserver(builder);
            CellIterator.Iterate(trunkId, adapter, observer);
            return builder.ToArray();
        }

        private static byte[] ReadWholeCell(InputAdapter adapter, int offset, int size)
        {
            Contract.Requires(adapter != null);
            Contract.Requires(size >= 0);
            Contract.Requires(offset >= 0);
            Contract.Requires(offset + size <= adapter.SharedBuffer.Length);
            
            var cell = new byte[size];
            Array.Copy(adapter.SharedBuffer, offset, cell, 0, size);
            return cell;
        }

        private bool CheckAllocation(long cellId)
        {
            Contract.Ensures(this.storage.Length > cellId);

            var expand = cellId >= this.storage.Length;
            if (expand)
            {
                Thread.MemoryBarrier();
                if (Interlocked.CompareExchange(ref this.lockValue, 1, 0) != 0)
                {
                    var wait = new SpinWait();
                    while (Interlocked.CompareExchange(ref this.lockValue, 1, 0) != 0)
                    {
                        wait.SpinOnce();
                    }
                }

                try
                {
                    expand = cellId >= this.storage.Length;
                    if (expand)
                    {
                        this.storage.Expand(cellId + PaddingSize);
                    }
                }
                finally
                {
                    this.lockValue = 0;
                }
            }

            Contract.Assume(this.storage.Length > cellId);
            return expand;
        }

        [ContractInvariantMethod]
        private void ObjectInvariant()
        {
            Contract.Invariant(this.storage != null);
        }
    }
}